package com.seaboxdata.interfaceJob;

import com.alibaba.fastjson.JSONObject;
import com.robin.core.base.util.IOUtils;
import com.seaboxdata.entity.BookingStat;
import com.seaboxdata.entity.VehicleStat;
import com.seaboxdata.entity.VehicleTypeStat;
import com.seaboxdata.service.BookingStatService;
import com.seaboxdata.service.VehicleStatService;
import com.seaboxdata.service.VehicleTypeStatService;
import com.seaboxdata.utils.DateUtil;
import com.seaboxdata.utils.ZipUtil;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.OutputStream;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.ftp.FTPFileSystem;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

/**
 * @author
 * @create 2021-04-06 11:34
 **/
@Component
@Configuration
@Slf4j
public class DpjtjFileCollectJob {

    private static final boolean isWin = System.getProperty("os.name").toLowerCase().contains("win");

    @Value("${dpjtj.ftp.host}")
    private String host;

    @Value("${dpjtj.ftp.port}")
    private String port;

    @Value("${dpjtj.ftp.username}")
    private String username;

    @Value("${dpjtj.ftp.password}")
    private String password;

    @Value("${dpjtj.unzip.password}")
    private String unzipPassword;

    @Value("${dpjtj.baseFilePath}")
    private String baseFilePath;

    @Value("${dpjtj.baseFile}")
    private String basePath;

    @Value("${dpjtj.srcFile}")
    private String srcPath;

    @Value("${dpjtj.dp.yysjtjb}")
    private String yysjtjb;

    @Value("${dpjtj.dp.gcfxtjb}")
    private String gcfxtjb;

    @Value("${dpjtj.dp.kkgcsjtjb}")
    private String kkgcsjtjb;

    @Value("${interface.isTest}")
    private String isTest;

    @Autowired
    private BookingStatService bookingStatService;

    @Autowired
    private VehicleTypeStatService vehicleTypeStatService;

    @Autowired
    private VehicleStatService vehicleStatService;

    @Scheduled(cron = "${dpjtj.interval}")
    public void doCollect() throws Exception {
        log.info("collect DpjtjFileCollectJob begin ");
        //如果是测试,直接退出
        if ("true".equals(isTest)) {
            return;
        }
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        String date1 = sdf.format(DateUtils.addDays(new Date(), -1));
        String date2 = sdf.format(DateUtils.addDays(new Date(), -2));
        handler(date1);
        copyFiles(date1);
        deleteFiles(date2);
        log.info("collect DpjtjFileCollectJob end");
    }

    /**
     * 从源FTP读取文件并解压后读取 TXT内容入库
     *
     * @param date
     * @throws Exception
     */
    public void handler(String date) throws Exception {
        if (isWin) {
            this.baseFilePath = "file:///D:/file";
            this.basePath = "D:\\file";
            this.srcPath = "D:\\file\\old";
        }
        FTPFileSystem fsLocal = new FTPFileSystem();
        org.apache.hadoop.conf.Configuration ftpConf = new org.apache.hadoop.conf.Configuration();
        ftpConf.set("fs.ftp.host", host);
        ftpConf.set("fs.ftp.host.port", port);
        ftpConf.set("fs.ftp.password." + host, password);
        ftpConf.set("fs.ftp.user." + host, username);

        FSDataInputStream in = null;
        FileReader fileReader = null;
        FileReader vehicleFileReader = null;
        FileReader vehicleStatsFileReader = null;
        OutputStream out = null;
        FileSystem fs = null;
        BufferedReader reader = null;
        BufferedReader vehicleTypeStatReader = null;
        BufferedReader vehicleStatsReader = null;
        try {
            log.info(" host {} ,port {}, ftpconf -> {}", host, port, ftpConf);
            fsLocal.initialize(URI.create("ftp//" + host + ":" + port), ftpConf);
            RemoteIterator<LocatedFileStatus> files = fsLocal.listFiles(new Path("/" + date), true);
            while (files.hasNext()) {
                log.info("循环读文件开始 -> {}", DateUtil.now());
                LocatedFileStatus next = files.next();
                Path path = next.getPath();
                log.info("next -> {} path -> {}", next, path);
                try {
                    in = fsLocal.open(path);
                } catch (Exception e1) {
                    e1.printStackTrace();
                    continue;
                }
                org.apache.hadoop.conf.Configuration localConf = new org.apache.hadoop.conf.Configuration();
                fs = FileSystem.get(URI.create(baseFilePath), localConf);
                out = fs.create(new Path(baseFilePath + path.toString()));
                IOUtils.copyBytes(in, out, 4096, true);
                ZipUtil.unZip(basePath + path.toString(),
                        basePath + path.toString().substring(0, path.toString().length() - 4), unzipPassword);
                //分别处理当天的各个文件的表数据

                List<BookingStat> bookingStats = new ArrayList<BookingStat>();
                File f = new File(
                        basePath + path.toString().substring(0, path.toString().length() - 4) + "/" + yysjtjb + ".txt");
                fileReader = new FileReader(f);
                reader = new BufferedReader(fileReader);
                reader.readLine();
                boolean flag = true;
                while (flag) {
                    String value = reader.readLine();
                    if (value == null) {
                        flag = false;
                    } else {
                        bookingStats.add(JSONObject.parseObject(value, BookingStat.class));
                    }
                }
                if (!CollectionUtils.isEmpty(bookingStats)) {
                    //入库逻辑
                    bookingStatService.saveBatch(bookingStats);
                }
                List<VehicleTypeStat> vehicleTypeStats = new ArrayList<VehicleTypeStat>();
                File fVehicleTypeStat = new File(
                        basePath + path.toString().substring(0, path.toString().length() - 4) + "/" + gcfxtjb + ".txt");
                vehicleFileReader = new FileReader(fVehicleTypeStat);
                vehicleTypeStatReader = new BufferedReader(vehicleFileReader);
                //跳过第一行,第一行是记录总条数,并无作用
                vehicleTypeStatReader.readLine();
                boolean flagVehicleTypeStat = true;
                while (flagVehicleTypeStat) {
                    String value = vehicleTypeStatReader.readLine();
                    if (value == null) {
                        flagVehicleTypeStat = false;
                    } else {
                        vehicleTypeStats.add(JSONObject.parseObject(value, VehicleTypeStat.class));
                    }
                }
                if (!CollectionUtils.isEmpty(vehicleTypeStats)) {
                    //入库逻辑
                    vehicleTypeStatService.saveBatch(vehicleTypeStats);
                }
                List<VehicleStat> vehicleStats = new ArrayList<VehicleStat>();
                File fVehicleStats = new File(
                        basePath + path.toString().substring(0, path.toString().length() - 4) + "/" + kkgcsjtjb
                                + ".txt");
                vehicleStatsFileReader = new FileReader(fVehicleStats);
                vehicleStatsReader = new BufferedReader(vehicleStatsFileReader);
                vehicleStatsReader.readLine();
                boolean flagVehicleStats = true;
                while (flagVehicleStats) {
                    String value = vehicleStatsReader.readLine();
                    if (value == null) {
                        flagVehicleStats = false;
                    } else {
                        vehicleStats.add(JSONObject.parseObject(value, VehicleStat.class));
                    }
                }
                if (!CollectionUtils.isEmpty(vehicleStats)) {
                    //入库逻辑
                    vehicleStatService.saveBatch(vehicleStats);
                }
                log.info("循环读文件结束 ->{}", DateUtil.now());
            }
            if (in != null) {
                in.close();
            }
            if (fileReader != null) {
                fileReader.close();
            }
            if (vehicleFileReader != null) {
                vehicleFileReader.close();
            }
            if (vehicleStatsFileReader != null) {
                vehicleStatsFileReader.close();
            }
            if (out != null) {
                out.close();
            }
            if (fs != null) {
                fs.close();
            }
            if (reader != null) {
                reader.close();
            }
            if (vehicleTypeStatReader != null) {
                vehicleTypeStatReader.close();
            }
            if (vehicleStatsReader != null) {
                vehicleStatsReader.close();
            }
            if (fsLocal != null) {
                fsLocal.close();
            }
            log.info("文件入库完成  -> end {}");
        } catch (Exception e) {
            e.printStackTrace();
            log.error("collect DpjtjFileCollectJob failed:" + e.getMessage());
        }
    }

    /**
     * 备份昨天的文件
     *
     * @param date
     */
    public void copyFiles(String date) throws Exception {
        File parentFile = new File(this.basePath + "/" + date);
        File srcFile = new File(srcPath + "/" + date);
        if (!srcFile.exists()) {
            srcFile.mkdir();
        }
        log.info("准备COPY 文件 -> start {}", DateUtil.now());
        copy(parentFile, srcFile);
        log.info("文件COPY 完成 -> end {}", DateUtil.now());
    }


    /**
     * COPY
     *
     * @param srcFile 源地址
     * @param distFile 目标地址
     * @throws Exception
     */
    public void copy(File srcFile, File distFile) throws Exception {
        log.info("正在备份文件 -> start{}", DateUtil.now());
        if (srcFile != null && srcFile.isFile()) {
            FileUtils.copyFile(srcFile, distFile, true);
        } else if (srcFile != null && srcFile.isDirectory()) {
            if (!srcFile.exists()) {
                srcFile.mkdir();
            }
            for (File file : srcFile.listFiles()) {
                File copyFlie = new File(distFile.getAbsolutePath() + "/" + file.getName());
                copy(file, copyFlie);
            }
        } else {
            log.info("源文件不存在 源地址-> {}" + srcFile);
        }
        log.info("正在备份文件 -> end {}", DateUtil.now());
    }


    /**
     * 删除文件,删除前天的文件
     *
     * @param date
     */
    public void deleteFiles(String date) throws Exception {
        File parentFile = new File(this.basePath + "/" + date);
        FileUtil.fullyDelete(parentFile, true);
        log.info("文件删除 完成 -> end {}", DateUtil.now());
    }
}

